package com.zyx.flinkdemo.sql.basic;

import com.zyx.flinkdemo.sql.cons.CommonConfig;
import com.zyx.flinkdemo.sql.utils.ConnectUtils;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import javax.naming.CompositeName;

/**
 * @author zyx
 * @since 2021/5/29 08:00
 * desc: 使用Flink将Kaka数据导入到MySQL中
 */
public class KafkaToMysqlDemo {
    public static void main(String[] args) throws Exception {
        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String fields = "id string," +
                "ts bigint," +
                "vc integer";
        // 注册Kafka的Source数据流表
        tableEnv.executeSql(ConnectUtils.getNormalKafkaSourceConnect("sensor_source", fields,
                "sensor_source", CommonConfig.KAFKA_SERVER,
                "src_consumer", "latest-offset", "csv"));
        // 注册MySQL的Sink表, 在MySQL中该表不会自动创建
        tableEnv.executeSql(ConnectUtils.getNormalMysqlSinkConnect("sensor_sink", fields, "linux201",
                "3306", "dbtest", "watersensor", "root", "123321"));
        // 使用Sql语句将Kafka中的数据插入到MySQL中
        tableEnv.executeSql("insert into sensor_sink select * from sensor_source");
    }
}
